/**
 * @file mqtt_client.c
 * @brief MQTT客户端实现 - IoT平台通信模块
 * @details 实现MQTT协议的订阅、发布等基础功能，基于Paho MQTT C库
 * 
 * @author zyb
 * @date 2025-10-30
 * @version 1.0
 */

#include "../include/mqtt_client.h"
#include "../../../common/logs/include/logger.h"
#include "../../../third/mqtt/src/MQTTClient.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <unistd.h>

// 日志模块名称
#define LOG_MODULE_NAME "MQTT_CLIENT"

// Paho MQTT超时时间
#define PAHO_TIMEOUT 10000L  // 10秒

/* ==================== MQTT客户端私有数据结构 ==================== */

/**
 * @brief MQTT客户端私有数据
 */
typedef struct mqtt_client {
    mqtt_config_t config;               /* 配置信息 */
    mqtt_status_t status;               /* 连接状态 */
    mqtt_stats_t stats;                 /* 统计信息 */
    
    /* 回调函数 */
    mqtt_message_callback_t message_callback;
    void* message_user_data;
    mqtt_status_callback_t status_callback;
    void* status_user_data;
    mqtt_error_callback_t error_callback;
    void* error_user_data;
    
    /* 线程相关 */
    pthread_mutex_t mutex;              /* 互斥锁 */
    
    /* Paho MQTT句柄 */
    MQTTClient paho_client;             /* Paho MQTT客户端句柄 */
} mqtt_client_t;

/* ==================== 全局变量 ==================== */

static bool g_mqtt_initialized = false;

/* ==================== 内部函数声明 ==================== */

/* Paho MQTT回调函数 */
static void paho_connection_lost(void* context, char* cause);
static int paho_message_arrived(void* context, char* topicName, int topicLen, MQTTClient_message* message);
static void paho_delivery_complete(void* context, MQTTClient_deliveryToken dt);

/* ==================== 函数实现 ==================== */

/**
 * @brief 初始化MQTT模块
 */
mqtt_result_t mqtt_init(void) {
    if (g_mqtt_initialized) {
        LOG_WARN(LOG_MODULE_NAME, "MQTT模块已经初始化");
        return MQTT_SUCCESS;
    }
    
    LOG_INFO(LOG_MODULE_NAME, "初始化MQTT模块(Paho MQTT C)");
    
    // 初始化Paho MQTT库（OpenSSL由应用程序自行处理）
    MQTTClient_init_options init_opts = MQTTClient_init_options_initializer;
    init_opts.do_openssl_init = 0;
    MQTTClient_global_init(&init_opts);
    
    g_mqtt_initialized = true;
    LOG_INFO(LOG_MODULE_NAME, "MQTT模块初始化成功");
    
    return MQTT_SUCCESS;
}

/**
 * @brief 清理MQTT模块
 */
mqtt_result_t mqtt_cleanup(void) {
    if (!g_mqtt_initialized) {
        return MQTT_SUCCESS;
    }
    
    LOG_INFO(LOG_MODULE_NAME, "清理MQTT模块");
    
    // Paho MQTT库的清理由系统自动处理，无需手动清理
    // OpenSSL由应用程序自行处理
    
    g_mqtt_initialized = false;
    LOG_INFO(LOG_MODULE_NAME, "MQTT模块清理完成");
    
    return MQTT_SUCCESS;
}

/**
 * @brief 创建MQTT客户端
 */
mqtt_client_handle_t mqtt_client_create(const mqtt_config_t* config) {
    if (config == NULL) {
        LOG_ERROR(LOG_MODULE_NAME, "配置参数为空");
        return NULL;
    }
    
    if (!g_mqtt_initialized) {
        LOG_ERROR(LOG_MODULE_NAME, "MQTT模块未初始化");
        return NULL;
    }
    
    LOG_DEBUG(LOG_MODULE_NAME, "开始创建MQTT客户端");
    LOG_DEBUG(LOG_MODULE_NAME, "配置信息 - Broker: %s, Port: %u", config->broker, config->port);
    LOG_DEBUG(LOG_MODULE_NAME, "配置信息 - ClientID: %s, Username: %s", 
              config->client_id, config->username[0] ? config->username : "(无)");
    LOG_DEBUG(LOG_MODULE_NAME, "配置信息 - KeepAlive: %u, QoS: %d, CleanSession: %s, TLS: %s",
              config->keepalive, config->qos, 
              config->clean_session ? "是" : "否",
              config->use_tls ? "是" : "否");
    
    // 分配内存
    mqtt_client_t* client = (mqtt_client_t*)malloc(sizeof(mqtt_client_t));
    if (client == NULL) {
        LOG_ERROR(LOG_MODULE_NAME, "内存分配失败");
        return NULL;
    }
    LOG_DEBUG(LOG_MODULE_NAME, "客户端内存分配成功: %zu 字节", sizeof(mqtt_client_t));
    
    // 初始化客户端结构
    memset(client, 0, sizeof(mqtt_client_t));
    memcpy(&client->config, config, sizeof(mqtt_config_t));
    client->status = MQTT_STATUS_DISCONNECTED;
    LOG_DEBUG(LOG_MODULE_NAME, "客户端结构初始化完成");
    
    // 初始化互斥锁
    if (pthread_mutex_init(&client->mutex, NULL) != 0) {
        LOG_ERROR(LOG_MODULE_NAME, "互斥锁初始化失败");
        free(client);
        return NULL;
    }
    LOG_DEBUG(LOG_MODULE_NAME, "互斥锁初始化成功");
    
    // 构造Broker URI
    char server_uri[512];
    if (config->use_tls) {
        snprintf(server_uri, sizeof(server_uri), "ssl://%s:%u", config->broker, config->port);
        LOG_DEBUG(LOG_MODULE_NAME, "使用SSL/TLS连接，URI: %s", server_uri);
    } else {
        snprintf(server_uri, sizeof(server_uri), "tcp://%s:%u", config->broker, config->port);
        LOG_DEBUG(LOG_MODULE_NAME, "使用TCP连接，URI: %s", server_uri);
    }
    
    // 创建Paho MQTT客户端
    LOG_DEBUG(LOG_MODULE_NAME, "调用MQTTClient_create创建Paho客户端");
    int rc = MQTTClient_create(&client->paho_client, server_uri, config->client_id,
                               MQTTCLIENT_PERSISTENCE_NONE, NULL);
    if (rc != MQTTCLIENT_SUCCESS) {
        LOG_ERROR(LOG_MODULE_NAME, "创建Paho MQTT客户端失败，错误码: %d", rc);
        pthread_mutex_destroy(&client->mutex);
        free(client);
        return NULL;
    }
    LOG_DEBUG(LOG_MODULE_NAME, "Paho MQTT客户端创建成功");
    
    // 设置回调函数
    LOG_DEBUG(LOG_MODULE_NAME, "设置MQTT回调函数");
    rc = MQTTClient_setCallbacks(client->paho_client, client, 
                                 paho_connection_lost, 
                                 paho_message_arrived, 
                                 paho_delivery_complete);
    if (rc != MQTTCLIENT_SUCCESS) {
        LOG_ERROR(LOG_MODULE_NAME, "设置Paho回调函数失败，错误码: %d", rc);
        MQTTClient_destroy(&client->paho_client);
        pthread_mutex_destroy(&client->mutex);
        free(client);
        return NULL;
    }
    LOG_DEBUG(LOG_MODULE_NAME, "回调函数设置成功");
    
    LOG_INFO(LOG_MODULE_NAME, "MQTT客户端创建[Broker: %s, ClientID: %s]", 
             server_uri, config->client_id);
    
    return (mqtt_client_handle_t)client;
}

/**
 * @brief 销毁MQTT客户端
 */
mqtt_result_t mqtt_client_destroy(mqtt_client_handle_t client) {
    if (client == NULL) {
        return MQTT_ERROR_INVALID_PARAM;
    }
    
    mqtt_client_t* mqtt_client = (mqtt_client_t*)client;
    
    LOG_INFO(LOG_MODULE_NAME, "销毁MQTT客户端");
    
    // 断开连接
    if (mqtt_client->status == MQTT_STATUS_CONNECTED) {
        mqtt_client_disconnect(client);
    }
    
    // 销毁Paho MQTT客户端
    MQTTClient_destroy(&mqtt_client->paho_client);
    
    // 销毁互斥锁
    pthread_mutex_destroy(&mqtt_client->mutex);
    
    // 释放内存
    free(mqtt_client);
    
    LOG_INFO(LOG_MODULE_NAME, "MQTT客户端销毁完成");
    
    return MQTT_SUCCESS;
}

/**
 * @brief 连接到MQTT Broker
 */
mqtt_result_t mqtt_client_connect(mqtt_client_handle_t client) {
    if (client == NULL) {
        return MQTT_ERROR_INVALID_PARAM;
    }
    
    mqtt_client_t* mqtt_client = (mqtt_client_t*)client;
    
    pthread_mutex_lock(&mqtt_client->mutex);
    
    if (mqtt_client->status == MQTT_STATUS_CONNECTED) {
        LOG_WARN(LOG_MODULE_NAME, "MQTT客户端已连接");
        pthread_mutex_unlock(&mqtt_client->mutex);
        return MQTT_SUCCESS;
    }
    
    LOG_INFO(LOG_MODULE_NAME, "正在连接到MQTT Broker: %s:%u", 
             mqtt_client->config.broker, mqtt_client->config.port);
    LOG_DEBUG(LOG_MODULE_NAME, "当前状态: %d, 准备连接到服务器", mqtt_client->status);
    
    mqtt_client->status = MQTT_STATUS_CONNECTING;
    
    // 配置连接选项
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    conn_opts.keepAliveInterval = mqtt_client->config.keepalive;
    conn_opts.cleansession = mqtt_client->config.clean_session ? 1 : 0;
    
    LOG_DEBUG(LOG_MODULE_NAME, "连接选项配置 - KeepAlive: %u秒, CleanSession: %s",
              conn_opts.keepAliveInterval, conn_opts.cleansession ? "是" : "否");
    
    // 设置用户名和密码（如果有）
    if (mqtt_client->config.username[0] != '\0') {
        conn_opts.username = mqtt_client->config.username;
        LOG_DEBUG(LOG_MODULE_NAME, "设置用户名: %s", mqtt_client->config.username);
    } else {
        LOG_DEBUG(LOG_MODULE_NAME, "未设置用户名（匿名连接）");
    }
    
    if (mqtt_client->config.password[0] != '\0') {
        conn_opts.password = mqtt_client->config.password;
        LOG_DEBUG(LOG_MODULE_NAME, "设置密码: %s", "******");
    } else {
        LOG_DEBUG(LOG_MODULE_NAME, "未设置密码");
    }
    
    // 构造完整的连接URI用于日志
    char server_uri[512];
    if (mqtt_client->config.use_tls) {
        snprintf(server_uri, sizeof(server_uri), "ssl://%s:%u", 
                 mqtt_client->config.broker, mqtt_client->config.port);
    } else {
        snprintf(server_uri, sizeof(server_uri), "tcp://%s:%u", 
                 mqtt_client->config.broker, mqtt_client->config.port);
    }
    LOG_DEBUG(LOG_MODULE_NAME, "连接URI: %s", server_uri);
    LOG_DEBUG(LOG_MODULE_NAME, "客户端ID: %s", mqtt_client->config.client_id);
    
    // 连接到Broker
    LOG_DEBUG(LOG_MODULE_NAME, "调用MQTTClient_connect开始连接...");
    int rc = MQTTClient_connect(mqtt_client->paho_client, &conn_opts);
    if (rc != MQTTCLIENT_SUCCESS) {
        LOG_ERROR(LOG_MODULE_NAME, "连接Broker失败，错误码: %d", rc);
        LOG_INFO(LOG_MODULE_NAME, "MQTT连接状态: DISCONNECTED");
        LOG_DEBUG(LOG_MODULE_NAME, "连接失败详情 - Broker: %s:%u, ClientID: %s, 用户名: %s",
                  mqtt_client->config.broker, mqtt_client->config.port,
                  mqtt_client->config.client_id,
                  mqtt_client->config.username[0] ? mqtt_client->config.username : "(无)");
        mqtt_client->status = MQTT_STATUS_ERROR;
        mqtt_client->stats.error_count++;
        pthread_mutex_unlock(&mqtt_client->mutex);
        return MQTT_ERROR_CONNECT_FAILED;
    }
    
    LOG_DEBUG(LOG_MODULE_NAME, "MQTTClient_connect返回成功，连接已建立");
    
    mqtt_client->status = MQTT_STATUS_CONNECTED;
    mqtt_client->stats.connection_count++;
    LOG_DEBUG(LOG_MODULE_NAME, "状态更新为CONNECTED，连接计数: %u", mqtt_client->stats.connection_count);
    
    // 保存回调函数和用户数据（避免在回调中使用已释放的锁）
    mqtt_status_callback_t status_callback = mqtt_client->status_callback;
    void* status_user_data = mqtt_client->status_user_data;
    
    // 先释放互斥锁，避免在回调函数中再次获取锁导致死锁
    pthread_mutex_unlock(&mqtt_client->mutex);
    
    // 在锁外调用状态回调，避免死锁
    if (status_callback) {
        LOG_DEBUG(LOG_MODULE_NAME, "调用状态回调函数通知连接成功（在锁外调用，避免死锁）");
        status_callback(MQTT_STATUS_CONNECTED, status_user_data);
    }
    
    LOG_INFO(LOG_MODULE_NAME, "MQTT连接成功 [%s:%u]", 
             mqtt_client->config.broker, mqtt_client->config.port);
    LOG_INFO(LOG_MODULE_NAME, "MQTT连接状态: CONNECTED");
    
    return MQTT_SUCCESS;
}

/**
 * @brief 断开MQTT连接
 */
mqtt_result_t mqtt_client_disconnect(mqtt_client_handle_t client) {
    if (client == NULL) {
        return MQTT_ERROR_INVALID_PARAM;
    }
    
    mqtt_client_t* mqtt_client = (mqtt_client_t*)client;
    
    pthread_mutex_lock(&mqtt_client->mutex);
    
    if (mqtt_client->status == MQTT_STATUS_DISCONNECTED) {
        pthread_mutex_unlock(&mqtt_client->mutex);
        return MQTT_SUCCESS;
    }
    
    LOG_INFO(LOG_MODULE_NAME, "断开MQTT连接");
    
    mqtt_client->status = MQTT_STATUS_DISCONNECTING;
    
    // 断开Paho MQTT连接
    int rc = MQTTClient_disconnect(mqtt_client->paho_client, PAHO_TIMEOUT);
    if (rc != MQTTCLIENT_SUCCESS) {
        LOG_WARN(LOG_MODULE_NAME, "断开连接失败: %d", rc);
    }
    
    mqtt_client->status = MQTT_STATUS_DISCONNECTED;
    
    // 保存回调函数和用户数据（避免在回调中使用已释放的锁）
    mqtt_status_callback_t status_callback = mqtt_client->status_callback;
    void* status_user_data = mqtt_client->status_user_data;
    
    // 先释放互斥锁，避免在回调函数中再次获取锁导致死锁
    pthread_mutex_unlock(&mqtt_client->mutex);
    
    // 在锁外调用状态回调，避免死锁
    if (status_callback) {
        status_callback(MQTT_STATUS_DISCONNECTED, status_user_data);
    }
    
    LOG_INFO(LOG_MODULE_NAME, "MQTT连接已断开");
    LOG_INFO(LOG_MODULE_NAME, "MQTT连接状态: DISCONNECTED");
    
    return MQTT_SUCCESS;
}

/**
 * @brief 订阅主题
 */
mqtt_result_t mqtt_client_subscribe(mqtt_client_handle_t client, 
                                    const char* topic, 
                                    mqtt_qos_t qos) {
    if (client == NULL || topic == NULL) {
        return MQTT_ERROR_INVALID_PARAM;
    }
    
    mqtt_client_t* mqtt_client = (mqtt_client_t*)client;
    
    pthread_mutex_lock(&mqtt_client->mutex);
    
    if (mqtt_client->status != MQTT_STATUS_CONNECTED) {
        LOG_ERROR(LOG_MODULE_NAME, "MQTT未连接");
        pthread_mutex_unlock(&mqtt_client->mutex);
        return MQTT_ERROR_NOT_CONNECTED;
    }
    
    LOG_DEBUG(LOG_MODULE_NAME, "准备订阅主题: %s (QoS=%d)", topic, qos);
    
    // 订阅主题
    LOG_DEBUG(LOG_MODULE_NAME, "调用MQTTClient_subscribe订阅主题");
    int rc = MQTTClient_subscribe(mqtt_client->paho_client, topic, qos);
    if (rc != MQTTCLIENT_SUCCESS) {
        LOG_ERROR(LOG_MODULE_NAME, "订阅主题失败，错误码: %d", rc);
        LOG_DEBUG(LOG_MODULE_NAME, "订阅失败详情 - 主题: %s, QoS: %d", topic, qos);
        mqtt_client->stats.error_count++;
        pthread_mutex_unlock(&mqtt_client->mutex);
        return MQTT_ERROR_SUBSCRIBE_FAILED;
    }
    
    pthread_mutex_unlock(&mqtt_client->mutex);
    
    LOG_INFO(LOG_MODULE_NAME, "主题订阅[%s]", topic);
    
    return MQTT_SUCCESS;
}

/**
 * @brief 取消订阅主题
 */
mqtt_result_t mqtt_client_unsubscribe(mqtt_client_handle_t client, 
                                      const char* topic) {
    if (client == NULL || topic == NULL) {
        return MQTT_ERROR_INVALID_PARAM;
    }
    
    mqtt_client_t* mqtt_client = (mqtt_client_t*)client;
    
    pthread_mutex_lock(&mqtt_client->mutex);
    
    if (mqtt_client->status != MQTT_STATUS_CONNECTED) {
        LOG_ERROR(LOG_MODULE_NAME, "MQTT未连接");
        pthread_mutex_unlock(&mqtt_client->mutex);
        return MQTT_ERROR_NOT_CONNECTED;
    }
    
    LOG_INFO(LOG_MODULE_NAME, "取消订阅主题: %s", topic);
    
    // 取消订阅主题
    int rc = MQTTClient_unsubscribe(mqtt_client->paho_client, topic);
    if (rc != MQTTCLIENT_SUCCESS) {
        LOG_ERROR(LOG_MODULE_NAME, "取消订阅失败: %d", rc);
        mqtt_client->stats.error_count++;
        pthread_mutex_unlock(&mqtt_client->mutex);
        return MQTT_ERROR_SUBSCRIBE_FAILED;
    }
    
    pthread_mutex_unlock(&mqtt_client->mutex);
    
    LOG_INFO(LOG_MODULE_NAME, "主题取消订阅成功: %s", topic);
    
    return MQTT_SUCCESS;
}

/**
 * @brief 发布消息
 */
mqtt_result_t mqtt_client_publish(mqtt_client_handle_t client,
                                  const char* topic,
                                  const uint8_t* payload,
                                  size_t payload_len,
                                  mqtt_qos_t qos,
                                  bool retained) {
    if (client == NULL || topic == NULL || payload == NULL) {
        LOG_DEBUG(LOG_MODULE_NAME, "发布消息参数校验失败");
        return MQTT_ERROR_INVALID_PARAM;
    }
    
    mqtt_client_t* mqtt_client = (mqtt_client_t*)client;
    
    pthread_mutex_lock(&mqtt_client->mutex);
    
    if (mqtt_client->status != MQTT_STATUS_CONNECTED) {
        LOG_ERROR(LOG_MODULE_NAME, "MQTT未连接，当前状态: %d", mqtt_client->status);
        pthread_mutex_unlock(&mqtt_client->mutex);
        return MQTT_ERROR_NOT_CONNECTED;
    }
    
    LOG_DEBUG(LOG_MODULE_NAME, "准备发布消息 - 主题: %s, 长度: %zu, QoS: %d, Retained: %s", 
              topic, payload_len, qos, retained ? "是" : "否");
    
    // 创建消息
    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    pubmsg.payload = (void*)payload;
    pubmsg.payloadlen = (int)payload_len;
    pubmsg.qos = qos;
    pubmsg.retained = retained ? 1 : 0;
    
    // 发布消息
    LOG_DEBUG(LOG_MODULE_NAME, "调用MQTTClient_publishMessage发布消息");
    MQTTClient_deliveryToken token;
    int rc = MQTTClient_publishMessage(mqtt_client->paho_client, topic, &pubmsg, &token);
    if (rc != MQTTCLIENT_SUCCESS) {
        LOG_ERROR(LOG_MODULE_NAME, "发布消息失败，错误码: %d", rc);
        LOG_DEBUG(LOG_MODULE_NAME, "发布失败详情 - 主题: %s, 长度: %zu", topic, payload_len);
        mqtt_client->stats.error_count++;
        pthread_mutex_unlock(&mqtt_client->mutex);
        return MQTT_ERROR_PUBLISH_FAILED;
    }
    LOG_DEBUG(LOG_MODULE_NAME, "消息发布调用成功，delivery token: %d", token);
    
    // 等待发送完成（QoS > 0时）
    if (qos > 0) {
        LOG_DEBUG(LOG_MODULE_NAME, "QoS=%d，等待消息发送完成确认...", qos);
        rc = MQTTClient_waitForCompletion(mqtt_client->paho_client, token, PAHO_TIMEOUT);
        if (rc != MQTTCLIENT_SUCCESS) {
            LOG_WARN(LOG_MODULE_NAME, "等待发送完成超时，错误码: %d", rc);
        } else {
            LOG_DEBUG(LOG_MODULE_NAME, "消息发送完成确认收到");
        }
    }
    
    // 更新统计
    mqtt_client->stats.messages_published++;
    mqtt_client->stats.bytes_sent += payload_len;
    LOG_DEBUG(LOG_MODULE_NAME, "统计更新 - 已发布: %u条, 已发送: %u字节", 
              mqtt_client->stats.messages_published, mqtt_client->stats.bytes_sent);
    
    pthread_mutex_unlock(&mqtt_client->mutex);
    
    LOG_DEBUG(LOG_MODULE_NAME, "消息发布流程完成");
    
    return MQTT_SUCCESS;
}

/**
 * @brief 获取连接状态
 */
mqtt_status_t mqtt_client_get_status(mqtt_client_handle_t client) {
    if (client == NULL) {
        return MQTT_STATUS_ERROR;
    }
    
    mqtt_client_t* mqtt_client = (mqtt_client_t*)client;
    
    pthread_mutex_lock(&mqtt_client->mutex);
    mqtt_status_t status = mqtt_client->status;
    pthread_mutex_unlock(&mqtt_client->mutex);
    
    return status;
}

/**
 * @brief 设置消息回调
 */
mqtt_result_t mqtt_client_set_message_callback(mqtt_client_handle_t client,
                                               mqtt_message_callback_t callback,
                                               void* user_data) {
    if (client == NULL) {
        return MQTT_ERROR_INVALID_PARAM;
    }
    
    mqtt_client_t* mqtt_client = (mqtt_client_t*)client;
    
    pthread_mutex_lock(&mqtt_client->mutex);
    mqtt_client->message_callback = callback;
    mqtt_client->message_user_data = user_data;
    pthread_mutex_unlock(&mqtt_client->mutex);
    
    LOG_DEBUG(LOG_MODULE_NAME, "消息回调设置成功");
    
    return MQTT_SUCCESS;
}

/**
 * @brief 设置状态回调
 */
mqtt_result_t mqtt_client_set_status_callback(mqtt_client_handle_t client,
                                              mqtt_status_callback_t callback,
                                              void* user_data) {
    if (client == NULL) {
        return MQTT_ERROR_INVALID_PARAM;
    }
    
    mqtt_client_t* mqtt_client = (mqtt_client_t*)client;
    
    pthread_mutex_lock(&mqtt_client->mutex);
    mqtt_client->status_callback = callback;
    mqtt_client->status_user_data = user_data;
    pthread_mutex_unlock(&mqtt_client->mutex);
    
    LOG_DEBUG(LOG_MODULE_NAME, "状态回调设置成功");
    
    return MQTT_SUCCESS;
}

/**
 * @brief 设置错误回调
 */
mqtt_result_t mqtt_client_set_error_callback(mqtt_client_handle_t client,
                                             mqtt_error_callback_t callback,
                                             void* user_data) {
    if (client == NULL) {
        return MQTT_ERROR_INVALID_PARAM;
    }
    
    mqtt_client_t* mqtt_client = (mqtt_client_t*)client;
    
    pthread_mutex_lock(&mqtt_client->mutex);
    mqtt_client->error_callback = callback;
    mqtt_client->error_user_data = user_data;
    pthread_mutex_unlock(&mqtt_client->mutex);
    
    LOG_DEBUG(LOG_MODULE_NAME, "错误回调设置成功");
    
    return MQTT_SUCCESS;
}

/**
 * @brief 获取统计信息
 */
mqtt_result_t mqtt_client_get_stats(mqtt_client_handle_t client, 
                                    mqtt_stats_t* stats) {
    if (client == NULL || stats == NULL) {
        return MQTT_ERROR_INVALID_PARAM;
    }
    
    mqtt_client_t* mqtt_client = (mqtt_client_t*)client;
    
    pthread_mutex_lock(&mqtt_client->mutex);
    memcpy(stats, &mqtt_client->stats, sizeof(mqtt_stats_t));
    pthread_mutex_unlock(&mqtt_client->mutex);
    
    return MQTT_SUCCESS;
}

/**
 * @brief 重置统计信息
 */
mqtt_result_t mqtt_client_reset_stats(mqtt_client_handle_t client) {
    if (client == NULL) {
        return MQTT_ERROR_INVALID_PARAM;
    }
    
    mqtt_client_t* mqtt_client = (mqtt_client_t*)client;
    
    pthread_mutex_lock(&mqtt_client->mutex);
    memset(&mqtt_client->stats, 0, sizeof(mqtt_stats_t));
    pthread_mutex_unlock(&mqtt_client->mutex);
    
    LOG_DEBUG(LOG_MODULE_NAME, "统计信息已重置");
    
    return MQTT_SUCCESS;
}

/**
 * @brief 检查连接状态
 */
bool mqtt_client_is_connected(mqtt_client_handle_t client) {
    if (client == NULL) {
        return false;
    }
    
    mqtt_client_t* mqtt_client = (mqtt_client_t*)client;
    pthread_mutex_lock(&mqtt_client->mutex);
    bool connected = (mqtt_client->status == MQTT_STATUS_CONNECTED);
    pthread_mutex_unlock(&mqtt_client->mutex);
    return connected;
}

/* ==================== 内部函数实现 - Paho MQTT回调 ==================== */

/**
 * @brief Paho MQTT连接丢失回调
 */
static void paho_connection_lost(void* context, char* cause) {
    mqtt_client_t* mqtt_client = (mqtt_client_t*)context;
    
    if (mqtt_client == NULL) {
        LOG_DEBUG(LOG_MODULE_NAME, "连接丢失回调收到空上下文");
        return;
    }
    
    const char* cause_str = cause ? cause : "未知原因";
    LOG_WARN(LOG_MODULE_NAME, "MQTT连接丢失: %s", cause_str);
    LOG_DEBUG(LOG_MODULE_NAME, "连接丢失详情 - Broker: %s:%u, 错误计数: %u",
              mqtt_client->config.broker, mqtt_client->config.port,
              mqtt_client->stats.error_count);
    
    pthread_mutex_lock(&mqtt_client->mutex);
    mqtt_client->status = MQTT_STATUS_DISCONNECTED;
    mqtt_client->stats.error_count++;
    LOG_DEBUG(LOG_MODULE_NAME, "状态更新为DISCONNECTED，错误计数: %u", mqtt_client->stats.error_count);
    pthread_mutex_unlock(&mqtt_client->mutex);
    
    // 调用状态回调
    if (mqtt_client->status_callback) {
        LOG_DEBUG(LOG_MODULE_NAME, "调用状态回调函数通知连接断开");
        mqtt_client->status_callback(MQTT_STATUS_DISCONNECTED, mqtt_client->status_user_data);
    }
    
    // 调用错误回调
    if (mqtt_client->error_callback) {
        LOG_DEBUG(LOG_MODULE_NAME, "调用错误回调函数");
        mqtt_client->error_callback(MQTT_ERROR_DISCONNECT_FAILED, mqtt_client->error_user_data);
    }
}

/**
 * @brief Paho MQTT消息到达回调
 */
static int paho_message_arrived(void* context, char* topicName, int topicLen, MQTTClient_message* message) {
    mqtt_client_t* mqtt_client = (mqtt_client_t*)context;
    
    if (mqtt_client == NULL || message == NULL) {
        LOG_DEBUG(LOG_MODULE_NAME, "消息到达回调收到空参数");
        return 1;
    }
    
    // 获取主题名称
    char* topic = topicName;
    if (topicLen > 0) {
        // 如果topicLen > 0，说明topic包含\0，需要特殊处理
        topic = (char*)malloc(topicLen + 1);
        if (topic) {
            memcpy(topic, topicName, topicLen);
            topic[topicLen] = '\0';
        }
    }
    
    LOG_DEBUG(LOG_MODULE_NAME, "收到MQTT消息 - 主题: %s, 长度: %d, QoS: %d", 
              topic ? topic : "NULL", message->payloadlen, message->qos);
    
    // 更新统计
    pthread_mutex_lock(&mqtt_client->mutex);
    mqtt_client->stats.messages_received++;
    mqtt_client->stats.bytes_received += message->payloadlen;
    LOG_DEBUG(LOG_MODULE_NAME, "统计更新 - 已接收: %u条, 已接收: %u字节",
              mqtt_client->stats.messages_received, mqtt_client->stats.bytes_received);
    pthread_mutex_unlock(&mqtt_client->mutex);
    
    // 调用消息回调
    if (mqtt_client->message_callback) {
        LOG_DEBUG(LOG_MODULE_NAME, "调用用户消息回调函数");
        mqtt_client->message_callback(topic ? topic : topicName, 
                                     (const uint8_t*)message->payload, 
                                     message->payloadlen, 
                                     mqtt_client->message_user_data);
    } else {
        LOG_DEBUG(LOG_MODULE_NAME, "未设置消息回调函数");
    }
    
    // 释放资源
    if (topic != topicName) {
        free(topic);
    }
    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    
    LOG_DEBUG(LOG_MODULE_NAME, "消息处理完成");
    return 1; // 返回1表示消息已处理
}

/**
 * @brief Paho MQTT消息发送完成回调
 */
static void paho_delivery_complete(void* context, MQTTClient_deliveryToken dt) {
    mqtt_client_t* mqtt_client = (mqtt_client_t*)context;
    
    if (mqtt_client == NULL) {
        return;
    }
    
    LOG_DEBUG(LOG_MODULE_NAME, "消息发送完成: token=%d", dt);
}

